/*!
gfs服务器的REST API
*/

use crate::*;
use anyhow::anyhow;
use axum::response::Result as AResult;
use axum::{
    body::{Body, Bytes},
    extract::{Query, State},
    http::{header, header::HeaderValue, Request},
    response::{IntoResponse, Json, Response},
};
use tower::ServiceExt;
use tower_http::services::ServeFile;
use std::collections::HashMap;

#[cfg(feature = "rotatelog")]
use log::{debug, error, warn};
#[cfg(feature = "tracelog")]
use tracing::{debug, error};

/**
axum 主程序入口
 */
pub(crate) async fn axum_main(shared: SharedHandle, port: u16) -> Result<()> {
    let assets_dir = PathBuf::from(&shared.gfs.root).join("static");
    let app = Router::new()
        .route("/", get(default_index))
        .route("/demo/form/file", get(demo_file_form).post(demo_form_file))
        .route("/file/:token/:name", post(gfs_stream_post))
        .route("/file/exist", get(gfs_file_exist))
        .route("/write/file/:token", post(gfs_form_write))
        .route("/record/:command", post(gfs_file_record))
        .route("/node/stats", get(gfs_read_stats))
        .route("/node/sync", get(gfs_distribute_sync))
        .route("/dir", get(gfs_dir))
        .route("/grep/file/:qry", get(gfs_grep_file))
        .route("/parallel/task", post(gfs_parallel_task))
        .route("/parallel/status/:taskname", get(gfs_parallel_status))
        .route("/search", get(gfs_search))
        .route("/whereis/:fname", get(gfs_whereis))
        .route("/download", get(gfs_download))
        .route("/reader", get(gfs_reader))
        .route("/image", get(gfs_image))
        .route("/admin/status/:token", get(gfs_admin_status))
        .nest_service(
            "/static",
            get_service(ServeDir::new(assets_dir)).handle_error(|error: io::Error| async move {
                (
                    StatusCode::INTERNAL_SERVER_ERROR,
                    format!("Unhandled internal error: {}", error),
                )
            }),
        )
        .with_state(Arc::clone(&shared))
        .layer(
            ServiceBuilder::new()
                // Handle errors from middleware
                .layer(HandleErrorLayer::new(handle_error))
                .load_shed()
                .concurrency_limit(1024)
                .timeout(Duration::from_secs(5))
                .layer(TraceLayer::new_for_http())
                .into_inner(),
        )
        .layer(DefaultBodyLimit::disable())
        .layer(RequestBodyLimitLayer::new(
            10 * 1024 * 1024, /* 250mb */
        ))
        .fallback(fallback);

    let addr = SocketAddr::from(([0, 0, 0, 0], port));
    debug!("listening on {}", addr);

    let r = axum::Server::bind(&addr)
        .serve(app.into_make_service())
        .with_graceful_shutdown(shutdown_signal("Axum web service"))
        .await;
    if let Err(err) = r {
        error!("{:#?}", err);
        bail!("create auxum web server failed")
    }
    Ok(())
}
/**
 * 访问不存在的uri时，显示的页面
 * - 为啥用fallback? 目前仅限于不存在的URI提示！！！
 */
async fn fallback(uri: Uri) -> impl IntoResponse {
    (
        StatusCode::NOT_FOUND,
        Html(format!(
            r#"<html><head><h1 align=center><Warnning</h1></head><body><BR><BR>
            <div align=center><span style="color:red">抓鱼去了 ～～ </span></div><BR>
            <div> No route for uri {}</div>
            </body>
            </html>"#,
            uri
        )),
    )
}
/**
handle出错时的展示页面
- 目前没有看到效果
- handle中抛出的Err，如果是MyError，则会被对应的Error实现的IntoResponse捕捉展现，比如NotLogin和GotoLogin。
- 如果是属于系统自身的Err，则直接应该走下面这个逻辑了（待验证确认）
*/
async fn handle_error(error: BoxError) -> impl IntoResponse {
    if error.is::<tower::timeout::error::Elapsed>() {
        return (
            StatusCode::REQUEST_TIMEOUT,
            Cow::from("request timed out  [ handle_error ]"),
        );
    }

    if error.is::<tower::load_shed::error::Overloaded>() {
        return (
            StatusCode::SERVICE_UNAVAILABLE,
            Cow::from("service is overloaded, try again later  [ handle_error ]"),
        );
    }

    (
        StatusCode::INTERNAL_SERVER_ERROR,
        Cow::from(format!(
            "!!!!!!!!!! Unhandled internal error: {} [ handle_error ]",
            error
        )),
    )
}

/**
REST API，借助ServFile的服务能力，提供文档的下载功能
 * http://localhost:9996/download?name=redis.conf&idx=17&token=2734234532975|123123
 * token具有时效性，防止盗链。
 */
pub async fn gfs_download(
    Query(para): Query<FileParam>,
    State(shared): State<SharedHandle>,
) -> AResult<Response, MyError> {
    if !verify_token(para.token.as_str(), TOKEN_TIMEOUT) {
        return AResult::Err(MyError::AnyErr(anyhow!("token is invalide")));
    }
    let filename =
        gfs_hashcode_to_path(&shared.gfs.root, para.ihash, para.idx).join(para.name.as_str());
    if !filename.exists() {
        debug!("file is not exist {:#?}", filename);
    }

    let save_file_header_value = format!("attachment;filename={}", para.name.as_str());
    let svc = ServeFile::new(filename);
    let mut res = svc.oneshot(Request::new(Body::empty())).await?;

    if let std::result::Result::Ok(header_value) = save_file_header_value.parse() {
        res.headers_mut()
            .append("Content-Disposition", header_value);
    }
    AResult::Ok(res.into_response())
}

/**
REST API，针对文本文件（或可提取文本的文件，如docx),提供的在线预览功能（无需下载，类似图片服务）
 * http://localhost:9996/reader?name=redis.conf&idx=17&token=2734234532975|123123
 * token具有时效性，防止盗链。
 * is_preview_file(ext) 可以判断是否可以预览
 */
pub async fn gfs_reader(
    Query(para): Query<FileParam>,
    State(shared): State<SharedHandle>,
) -> AResult<Response, MyError> {
    if !verify_token(para.token.as_str(), TOKEN_TIMEOUT) {
        return AResult::Err(MyError::AnyErr(anyhow!("token is invalide")));
    }
    // let ihash = string_hash256(para.name.as_str());
    let mut filename =
        gfs_hashcode_to_path(&shared.gfs.root, para.ihash, para.idx).join(para.name.as_str());

    if !filename.exists() {
        debug!("file is not exist {:#?}", filename);
        return AResult::Err(MyError::AnyErr(anyhow!("file is not exist!")));
    }
    // let txt: Vec<&str> = vec!["txt", "rs", "py", "go", "conf", "toml"];
    let fother = PathBuf::from(&filename);
    let mut ext = fother.extension().unwrap().to_str().unwrap();
    let extr: Vec<&str> = vec!["docx", "pptx"];
    if extr.contains(&ext) {
        let parent = if let Some(parent) = filename.parent() {
            PathBuf::from(parent)
        } else {
            PathBuf::from(&shared.gfs.root)
                .join(format!("{:02x}", para.ihash))
                .join(format!("{:02x}", para.idx))
        };
        let txtf = PathBuf::from(parent).join(format!("{}.rabbit.txt", para.idx));
        if txtf.exists() {
            filename = txtf;
            ext = "txt"
        }
    }

    if is_preview_file(ext) {
        let fsrc = String::from_utf8(
            std::fs::read(filename)
                .map_err(|_| anyhow!("read error"))?
                .to_vec(),
        )
        .map_err(|_| anyhow!("read error"))?;

        let headers = [(
            axum::http::header::CONTENT_TYPE,
            axum::http::HeaderValue::from_static("text/plain;charset=UTF-8"),
        )];
        return AResult::Ok((headers, fsrc).into_response());
    }
    AResult::Err(MyError::AnyErr(anyhow!("file format is miss")))
}

/**
REST API，借助ServFile的文件服务能力，针对图片文件（特定的扩展类别），提供在线图片服务
 * http://localhost:9996/image?name=abc.jpeg&idx=5&token=2101e1a2|1232112
 * token具有时效性，防止盗链。
 * is_image_file(ext) 可以判断是否可以提供在线服务图片
 */
pub async fn gfs_image(
    Query(para): Query<FileParam>,
    State(shared): State<SharedHandle>,
) -> AResult<Response, MyError> {
    if !verify_token(para.token.as_str(), TOKEN_TIMEOUT) {
        return AResult::Err(MyError::AnyErr(anyhow!("token is invalide")));
    }
    let ihash = string_hash256(para.name.as_str());
    let filename = gfs_hashcode_to_path(&shared.gfs.root, ihash, para.idx).join(para.name);
    if !filename.exists() {
        return AResult::Err(MyError::AnyErr(anyhow!("file is not exist!")));
    }
    // let images: Vec<&str> = vec!["jpg", "jpeg", "png", "svg"];
    let ext = filename
        .extension()
        .unwrap()
        .to_str()
        .unwrap()
        .to_lowercase();
    if !is_image_file(ext.as_str()) {
        // images.contains(&ext.as_str()) {
        return AResult::Err(MyError::AnyErr(anyhow!("type is not image file!")));
    }

    let svc = ServeFile::new(filename);
    let res = svc.oneshot(Request::new(Body::empty())).await?;
    AResult::Ok(
        (
            [(header::CONTENT_TYPE, HeaderValue::from_static("image/png"))],
            res,
        )
            .into_response(),
    )
}

/**
REST API, 流式写入文件
 * curl -- help upload
 * curl -T "xiao.png" -X POST http://localhost:9996/file/token/xiao.png
 * 这里借用了FileParam来解析路径参数，syntax: /file/:token/:name
 */
pub async fn gfs_stream_post(
    Path(para): Path<FileParam>,
    State(shared): State<SharedHandle>,
    body: BodyStream,
) -> AResult<Response, MyError> {
    // } -> std::result::Result<GfsResponse, (StatusCode, String)> {
    debug!("para = {:#?}", para);
    if !verify_token(para.token.as_str(), TOKEN_TIMEOUT) {
        return AResult::Err(MyError::AnyErr(anyhow!("token is invalide")));
    }

    if !shared.is_master() {
        return AResult::Err(MyError::AnyErr(anyhow!(
            "current is slave, can't write, readonly!"
        )));
    }

    let fhash = shared.get_next_id().await;
    let ret = stream_to_file(
        "should write to local",
        &shared.gfs.root,
        &para.name,
        fhash,
        body,
    )
    .await;
    match ret {
        std::result::Result::Ok(mut ret) => {
            ret.idx = shared
                .gfs
                .add_stat(ret.ihash, ret.idx, para.name.as_str())
                .await?;
            shared
                .tx
                .send(Task::New(ret.clone()))
                .await
                .or_else(|_| bail!("error to send"))?;

            AResult::Ok(Json(ret).into_response())
        }
        Err((_, err)) => {
            error!("stream post error: {:#?}", err);
            AResult::Err(MyError::AnyErr(anyhow!(err)))
        }
    }
}

/**
Record格式文件写API，指定路径（ihash, idx）和文件名，文本内容记录
 - 指定路径（ihash, idx）和文件名，跳过hash策略和stat记录，直接向文件中写入指定内容（ 如果当前是master，则文件会被同步到slave,否则忽略 ）
 - command， clear or override，则文件会被重置清空。否则就是append，追加模式。
 - ikey 如果 <0, 则被视为普通文件，否则将被视为任务记录文件，将维护任务执行状态，此时ikey表示ihash=ikey的目录节点的记录数据。
 - 不可与存档类文件冲突（ihash/dix/文件名 不可与存档类记录冲突 ）

 */
pub async fn gfs_file_record(
    Path(command): Path<String>,
    Query(para): Query<RecordParam>,
    State(shared): State<SharedHandle>,
    body: Bytes,
) -> AResult<Response, MyError> {
    if !verify_token(para.token.as_str(), TOKEN_TIMEOUT) {
        return AResult::Err(MyError::AnyErr(anyhow!("token is invalide")));
    }
    // error!(?command, ?para);
    /* record 类型文件，不能与存档类文件冲突，不可修改、覆盖存档类文件（但存档类可以反之）
     * 如果hash 一致，则查看指定idx下是否也有同名文件
     * 此时只能保证当前没有符合条件的存档类文件，但可能后续会有！此种情况下，存档类将会覆盖掉record文件
     *
     */
    let mut doit = true;
    let ihash = string_hash256(&para.name);
    if ihash == para.ihash {
        let stat = shared
            .get_by_name(
                &format!("{:02x}", para.ihash % 256),
                &para.name,
                para.idx,
                1,
            )
            .await?;
        if stat.len() > 0 && stat[0].id == para.idx {
            doit = false;
        }
    }
    if doit == false {
        return AResult::Err(MyError::AnyErr(anyhow!(
            "此文件名不合法，不可以以record格式写入此目录下！"
        )));
    }
    let _r = shared
        .tx
        .send(Task::Record(para, command, body))
        .await
        .or_else(|_| bail!("error to send"))?;
    AResult::Ok("OK".into_response())
}

/**
对master上特定文件（ihash，idx，name）进行node范围内同步（仅当开启MQ同步功能）
 - master_server/node/sync? with FileParam
 - 不做ihash与name的校验，所以只要文件存在，即可以同步（可以是gfs托管写入的文件，也可是gfs record类型文件）
 - 通常用于master写入record类型文件（不会自动同步），然后调用此函数，进行显式同步。
 */
pub async fn gfs_distribute_sync(
    Query(para): Query<FileParam>,
    State(shared): State<SharedHandle>,
) -> AResult<Response, MyError> {
    if !verify_token(para.token.as_str(), TOKEN_TIMEOUT) {
        return AResult::Err(MyError::AnyErr(anyhow!("token is invalide")));
    }

    if !shared.is_master() {
        return AResult::Err(MyError::AnyErr(anyhow!(
            "current is slave, can't distribute sync"
        )));
    }

    let filename = gfs_hashcode_to_path(&shared.gfs.root, para.ihash, para.idx).join(&para.name);
    if !filename.exists() {
        return AResult::Err(MyError::AnyErr(anyhow!("file is not exist!")));
    }
    /*
    如果文件名hash定位成功，则表示是归档类文件，code = 0;
    否则，将视为record类型的文件同步，code = SYNC_CODE_RECORD
    slave解析后，对于SYNC_CODE_RECORD，将理解为record类文件，存在异步冲突问题，需要加锁保护。

     */
    let ihash = string_hash256(&para.name);
    let code = if ihash == para.ihash {
        0
    } else {
        SYNC_CODE_RECORD
    };

    let todo = GfsResponse {
        name: para.name,
        ihash: para.ihash,
        idx: para.idx,
        code,
    };
    shared
        .tx
        .send(Task::New(todo))
        .await
        .or_else(|_| bail!("error to send"))?;

    AResult::Ok("OK".into_response())
}
/**
REST API，form形式写入文件
 * Handler that accepts a multipart form upload and streams each field to a file.
 * syntax: form/file/:token
 * curl -F "file=@xiao.png" -X POST http://localhost:9996/form/file/:token
 */
pub async fn gfs_form_write(
    Path(token): Path<String>,
    State(shared): State<SharedHandle>,
    mut multipart: Multipart,
) -> AResult<Response, MyError> {
    if !verify_token(token.as_str(), TOKEN_TIMEOUT) {
        return AResult::Err(MyError::AnyErr(anyhow!("token is invalide")));
    }
    if !shared.is_master() {
        return AResult::Err(MyError::AnyErr(anyhow!(
            "current is slave, can't write, readonly!"
        )));
    }

    let mut rets: Vec<GfsResponse> = vec![];
    while let Some(field) = multipart.next_field().await.unwrap() {
        let file_name = if let Some(file_name) = field.file_name() {
            file_name.to_owned()
        } else {
            continue;
        };
        debug!("filename = {}", file_name);
        let idx = shared.get_next_id().await;

        if let std::result::Result::Ok(mut ret) = stream_to_file(
            "should write local",
            &shared.gfs.root,
            &file_name,
            idx,
            field,
        )
        .await
        {
            ret.idx = shared
                .gfs
                .add_stat(ret.ihash, ret.idx, file_name.as_str())
                .await?;
            shared
                .tx
                .send(Task::New(ret.clone()))
                .await
                .or_else(|_| bail!("error to send"))?;
            rets.push(ret);
        }
    }
    AResult::Ok(Json(rets).into_response())
}

/**
REST API，获取当前node的stat信息（可以是master，也可以是slave，如果同步及时，应该都相同）
 * http://localhost:9996/node/stats?from=0&limit=10
 *
 */
pub async fn gfs_read_stats(
    Query(para): Query<GfsParams>,
    State(shared): State<SharedHandle>,
) -> AResult<Response, MyError> {
    if !verify_token(para.token.as_str(), TOKEN_TIMEOUT) {
        return AResult::Err(MyError::AnyErr(anyhow!("token is invalide")));
    }

    let retv = shared.gfs.get_next_from_id(para.from, para.limit).await?;
    AResult::Ok(Json(retv).into_response())
}

/**
REST API, 获取当前指定ihash目录下的文件信息，idx为起始最小id，limit为返回限制
 * http://localhost:9996/dir?ihash=1&idx=1&limit=20&token=1247129a723e38832se|323344
 *
 */
pub async fn gfs_dir(
    Query(para): Query<FileParam>,
    State(shared): State<SharedHandle>,
) -> AResult<Response, MyError> {
    debug!("para = {:#?}", para);
    if !verify_token(para.token.as_str(), TOKEN_TIMEOUT) {
        return AResult::Err(MyError::AnyErr(anyhow!("token is invalide")));
    }
    let limit: u32 = if para.limit > 0 { para.limit } else { 1024 };
    let retv = shared
        .gfs
        .get_dir(
            format!("{:02x}", para.ihash % 256).as_str(),
            para.idx,
            limit,
        )
        .await?;
    AResult::Ok(Json(retv).into_response())
}

/**
检测一个文件是否存在

 */
pub async fn gfs_file_exist(
    Query(para): Query<FileParam>,
    State(shared): State<SharedHandle>,
) -> AResult<Response, MyError> {
    if !verify_token(para.token.as_str(), TOKEN_TIMEOUT) {
        return AResult::Err(MyError::AnyErr(anyhow!("token is invalide")));
    }
    let filename =
        gfs_hashcode_to_path(&shared.gfs.root, para.ihash, para.idx).join(para.name.as_str());

    if !filename.exists() {
        debug!("file is not exist {:#?}", filename);
        return AResult::Err(MyError::AnyErr(anyhow!("file is not exist!")));
    }
    return AResult::Ok("OK".into_response());
}

/**
TODO， comming soon
 */
pub async fn gfs_admin_status(
    Path(token): Path<String>,
    Query(mode): Query<HashMap<String, String>>,
    State(shared): State<SharedHandle>,
) -> AResult<Response, MyError> {
    if !verify_token(token.as_str(), TOKEN_TIMEOUT) && !token.as_str().eq("admin***") {
        return AResult::Err(MyError::AnyErr(anyhow!("token is invalide")));
    }

    let slaves = if let Some(slaves) = shared.get_slaves().await{
        slaves
    }else {vec!["".to_string()] };

    let mut html = false;
    if let Some(mode) = mode.get("f") {
        if mode.eq("html") {
            html = true;
        }
    }

    let retv = shared.gfs.get_node_status().await?;
    let mq = &shared.mq_server;
    if html {
        let mut rets = String::from("<html><body><h2 align=center>RabbitGFS Server Status <h2><BR><div align=center><table border=1 cellspacing=0>");
        if shared.is_master() { // Master
            rets.push_str("<tr><td><b>Actor</b></td><td><b><span style='color:red'>Master</span></b></td></tr>");
            rets.push_str(&format!("<tr><td><b>Port</b></td><td>{}</td></tr>", shared.port));
            rets.push_str(&format!("<tr><td><b>MQ</b></td><td>{}</td></tr>", mq));
            rets.push_str(&format!("<tr><td><b>MQ-consumer</b></td><td>{}</td></tr>", false ));
            rets.push_str(&format!("<tr><td><b>Slaves counts</b></td><td>{:#?}</td></tr>", slaves.len() ));
            rets.push_str("</table></div>");

            if slaves.len() > 0 {
                rets.push_str("<BR><h3 align=center>Slave 服务器</h3>");
                rets.push_str("<div align=center><table border=1 cellspacing=0><tr><th>INODE</th><th> STATUS</th></tr>");        
           
                for i in 0..slaves.len() {
                    rets.push_str(&format!("<tr><td>Slave-{}</td><td><a href='http://{}/admin/status/admin***?f=html'>{}</a></td></tr>", i+1, &slaves[i], &slaves[i] ));                    
                }
                rets.push_str("</table></div>");    
            }            
           
            rets.push_str("<h3 align=center>当前(Master)服务器文件统计</h3>");
            rets.push_str("<div align=center><table border=1 cellspacing=0><tr><th>INODE</th><th>FILE COUNTS</th></tr>");            
            for i in 0..retv.len() {
                let (k,v) = &retv[i];
                rets.push_str(&format!("<tr><td>{}</td><td>{}</td></tr>", k, *v));    
            }            
            rets.push_str("</table></div>");                        

        } else {
            let mqc = if shared.mq_server.len() > 0 {
                true
            } else {
                false
            };
            let master = &shared.gfs.master;
            rets.push_str("<tr><td><b>Actor<b></td><td><b><span style='color:red'>Slave</span></b></td></tr>");
            rets.push_str(&format!("<tr><td><b>Port</b></td><td>{}</td></tr>", shared.port));
            rets.push_str(&format!("<tr><td><b>MQ</b></td><td>{}</td></tr>", mq));
            rets.push_str(&format!("<tr><td><b>Master<b></td><td><a href='{}/admin/status/admin***?f=html'>{}</a></td></tr>",master, master));
            rets.push_str(&format!("<tr><td><b>MQ-consumer<b></td><td>{}</td></tr>", true ));

            rets.push_str("</table></div><BR>");            
            rets.push_str("<h3 align=center>当前服务器文件统计</h3>");
            rets.push_str("<div align=center><table border=1 cellspacing=0><tr><th>INODE</th><th>FILE COUNTS</th></tr>");
            for i in 0..retv.len() {
                let (k,v) = &retv[i];
                rets.push_str(&format!("<tr><td>{}</td><td>{}</td></tr>", k, *v));    
            }             
            rets.push_str("</table></div>");                        
        }
        rets.push_str("</body></html>");            
        AResult::Ok(Html(rets).into_response())
    } else {
        if shared.is_master() {
            let data =
                serde_json::json!({"actor": "master","MQ":mq, "MQ-consumer": false ,"node":retv});
            AResult::Ok(Json(data).into_response())
        } else {
            let mqc = if shared.mq_server.len() > 0 {
                true
            } else {
                false
            };
            let master = &shared.gfs.master;
            let data = serde_json::json!({"actor": "slave", "master": master , "MQ":mq, "MQ-consumer": mqc ,"node":retv});
            AResult::Ok(Json(data).into_response())
        }
    }
}
